package com.zh.rocketmq;

import com.zh.mycommon.util.MyIdGenerator;
import com.zh.rocketmq.mqmsg.MqMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalTime;

/**
 * 消息生产者
 * <p>
 * 消息的发送方式分为：同步 / 异步 / 单向
 * <p>
 * 消息的消费时机分为：普通消息、延时消息
 * <p>
 * 消息的消费顺序分为：普通消息、顺序消息
 */
@Slf4j
@RestController
@RequestMapping("/producer")
public class ProducerController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    static final String topic = "DemoTopic";
    static final String tx_topic = "TxTopic";


    //同步消息
    // 同步发送消息是指，Producer 发出⼀条消息后，会在收到 MQ 返回的 ACK 之后才发下⼀条消息。
    // 该方式的消息可靠性最高，但消息发送效率太低。RocketMQ 同步消息的方法形如 syncXx()。
    @GetMapping("/sync")
    public String syncMessageSend(@RequestParam(value = "size", defaultValue = "1") Integer size) {
        log.info("【同步消息】 开始 ...");

        for (int i = 0; i < size; i++) {
            MqMsg mqMsg = new MqMsg(MyIdGenerator.simpleUuid(), "hello rocketmq - " + LocalTime.now(), i);

            log.info("sync send message ... data = {}", mqMsg);
            SendResult sendResult = rocketMQTemplate.syncSend(topic, mqMsg);

            log.info("sync message sent : result = {}", sendResult);
        }
        log.info("【同步消息】 结束。sync send {} count", size);
        return "success";
    }

    //异步消息
    // 异步发送消息是指，Producer 发出消息后无需等待 MQ 返回 ACK，直接发送下⼀条消息。
    // 该方式的消息可靠性可以得到保障，消息发送效率也可以。RocketMQ 同步消息的方法形如 asyncXx()
    @GetMapping("/async")
    public String asyncMessageSend(@RequestParam(value = "size", defaultValue = "1") Integer size) {
        log.info("【异步消息】 开始 ...");

        var callback = new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("async message sent : result = {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("async message error: {}", throwable.getMessage());
            }
        };

        for (int i = 0; i < size; i++) {
            MqMsg mqMsg = new MqMsg(MyIdGenerator.simpleUuid(), "hello rocketmq - " + LocalTime.now(), i);
            log.info("async send message ... data = {}", mqMsg);
            rocketMQTemplate.asyncSend(topic, mqMsg, callback);
        }
        log.info("【异步消息】 结束。async send {} count", size);
        return "success";
    }

    //单向消息
    // 单向发送消息是指，Producer 仅负责发送消息，不等待、不处理 MQ 的 ACK。
    // 该发送方式时 MQ 也不返回 ACK。该方式的消息发送效率最高，但消息可靠性较差
    @GetMapping("/oneway")
    public String onewayMessageSend(@RequestParam(value = "size", defaultValue = "1") Integer size) {
        log.info("【单向消息】 开始 ...");

        for (int i = 0; i < size; i++) {
            MqMsg mqMsg = new MqMsg(MyIdGenerator.simpleUuid(), "hello rocketmq - " + LocalTime.now(), i);

            log.info("oneway send message ... data = {}", mqMsg);
            rocketMQTemplate.sendOneWay(topic, mqMsg);
        }
        log.info("【单向消息】 结束。oneway send {} count", size);
        return "success";
    }

    /* ----------------------------------------------------------------------------------------------- */


    //延时消息（同步发送、异步发送、单向发送 都可以发送延时消息）
    // 延时消息指的是，当消息写入到Broker后，在指定的时长后才可被消费处理的消息。
    // RocketMQ 提供了 18 个延时等级DelayLevel，分别是：
    // 1-1s 2-5s 3-10s 4-30s 5-1m 6-2m 7-3m 8-4m 9-5m 10-6m 11-7m 12-8m 13-9m 14-10m 15-20m 16-30m 17-1h 18-2h
    @GetMapping("/delay")
    public String delayMessageSend(@RequestParam(value = "size", defaultValue = "1") Integer size) {
        log.info("【延时消息】 开始 ...");

        for (int i = 0; i < size; i++) {
            MqMsg mqMsg = new MqMsg(MyIdGenerator.simpleUuid(), "hello rocketmq - " + LocalTime.now(), i);

            log.info("delay send message ... data = {}", mqMsg);
            SendResult sendResult = rocketMQTemplate.syncSendDelayTimeSeconds(topic, mqMsg, 10);//10s
            //可以同步或异步发送延时消息
            //or rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(mqMsg).build(), callback, 1000, 3); //延迟10s

            log.info("delay message sent : result = {}", sendResult);
        }
        log.info("【延时消息】 结束。delay send {} count", size);
        return "success";
    }


    /* ----------------------------------------------------------------------------------------------- */


    //顺序消息（同步发送、异步发送、单向发送 都可以发送顺序消息）
    // 顺序消息指的是，严格按照消息的发送顺序进行消费的消息(FIFO)。
    @GetMapping("/orderly")
    public String sequenceMessageSend(@RequestParam(value = "size", defaultValue = "1") Integer size
    , @RequestParam(value = "groupby", defaultValue = "global") String groupby) {
        log.info("【顺序消息】 开始 ...");

        for (int i = 0; i < size; i++) {
            MqMsg mqMsg = new MqMsg(MyIdGenerator.simpleUuid(), "hello rocketmq - " + LocalTime.now(), i);

            log.info("orderly send message ... data = {} . groupby: {}", mqMsg, groupby);
            SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic, mqMsg, groupby);

            log.info("orderly message sent : result = {}", sendResult);
        }
        log.info("【顺序消息】 结束。orderly send {} count", size);
        return "success";
    }


    /* ----------------------------------------------------------------------------------------------- */


    //事务消息
    @GetMapping("/transaction")
    public String transactionMessageSend(@RequestParam(value = "size", defaultValue = "1") Integer size) {
        log.info("【事务消息】 开始 ...");


        //假设场景：1-创建订单（订单服务本地落库），2-发送库存扣减的MQ消息（库存服务消费消息）
        //事务MQ的发送流程：
        //1-发送事务消息到MQ（发送库存扣减的MQ半消息）
        //2-本地事务处理（订单服务本地落库）
        //3-根据本地事务处理结果，确认MQ消息是否提交或回滚
        for (int i = 0; i < size; i++) {
            MqMsg mqMsg = new MqMsg(MyIdGenerator.simpleUuid(), "hello rocketmq - " + LocalTime.now(), i);

            log.info("transaction send message ... data = {} ", mqMsg);
            TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction(tx_topic, MessageBuilder.withPayload(mqMsg).build(), null);

            log.info("transaction message sent : result = {}", sendResult);
        }
        log.info("【事务消息】 结束。transaction send {} count", size);
        return "success";
    }
}
